package io.jnet.server;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.AsynchronousChannelGroup;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import com.jfireframework.baseutil.concurrent.MPMCQueue;
import com.jfireframework.baseutil.exception.JustThrowException;
import io.jnet.server.AioServerBuilder.IoMode;
import io.jnet.server.accepthandler.ConcentrationAcceptHandler;
import io.jnet.server.accepthandler.SimpleAcceptHandler;
import io.jnet.server.accepthandler.TaskAttachAcceptHandler;
import io.jnet.server.api.AcceptHandler;
import io.jnet.server.api.ServerChannelContextBuilder;
import io.jnet.server.api.ServerListener;
import io.jnet.server.businessprocessor.ConcentrationProcessor;
import io.jnet.server.util.ProcessorTask;

public class AioServer
{
    private AsynchronousChannelGroup        channelGroup;
    private AsynchronousServerSocketChannel serverSocketChannel;
    private String                          ip;
    private int                             port;
    private AcceptHandler                   acceptHandler;
    private IoMode                          ioMode;
    private ServerChannelContextBuilder     serverChannelContextBuilder;
    private int                             businessProcessorNum;
    private ServerListener                  serverListener;
    
    AioServer(int businessProcessorNum, String ip, int port, AsynchronousChannelGroup channelGroup, IoMode ioMode, ServerChannelContextBuilder serverChannelContextBuilder, ServerListener serverListener)
    {
        this.businessProcessorNum = businessProcessorNum;
        this.ip = ip;
        this.port = port;
        this.channelGroup = channelGroup;
        this.ioMode = ioMode;
        this.serverChannelContextBuilder = serverChannelContextBuilder;
        this.serverListener = serverListener;
    }
    
    /**
     * 以端口初始化server服务器。
     * 
     * @param port
     */
    public void start()
    {
        try
        {
            serverSocketChannel = AsynchronousServerSocketChannel.open(channelGroup);
            serverSocketChannel.bind(new InetSocketAddress(ip, port), 0);
            switch (ioMode)
            {
                case simple:
                    acceptHandler = new SimpleAcceptHandler(serverSocketChannel, serverChannelContextBuilder, serverListener);
                    break;
                case concentration:
                {
                    ExecutorService executorService = Executors.newFixedThreadPool(businessProcessorNum, new ThreadFactory() {
                        int i = 1;
                        
                        @Override
                        public Thread newThread(Runnable r)
                        {
                            return new Thread(r, "服务端业务处理线程-" + (i++));
                        }
                    });
                    MPMCQueue<ProcessorTask> tasks = new MPMCQueue<>();
                    for (int i = 0; i < businessProcessorNum; i++)
                    {
                        executorService.execute(new ConcentrationProcessor(tasks, serverListener));
                    }
                    acceptHandler = new ConcentrationAcceptHandler(executorService, tasks, serverSocketChannel, serverChannelContextBuilder, serverListener);
                    break;
                }
                case TASK_ATTACH:
                {
                    ExecutorService executorService = Executors.newFixedThreadPool(businessProcessorNum, new ThreadFactory() {
                        int i = 1;
                        
                        @Override
                        public Thread newThread(Runnable r)
                        {
                            return new Thread(r, "服务端业务处理线程-" + (i++));
                        }
                    });
                    acceptHandler = new TaskAttachAcceptHandler(executorService, serverSocketChannel, serverChannelContextBuilder, serverListener);
                    break;
                }
                default:
                    break;
            }
            serverSocketChannel.accept(null, acceptHandler);
        }
        catch (IOException e)
        {
            throw new JustThrowException(e);
        }
    }
    
    public void stop()
    {
        try
        {
            acceptHandler.stop();
            serverSocketChannel.close();
            channelGroup.shutdownNow();
            channelGroup.awaitTermination(10, TimeUnit.SECONDS);
        }
        catch (Exception e)
        {
            throw new RuntimeException(e);
        }
    }
}
